package com.vteba.utils.serialize;

import java.util.List;
import java.util.NoSuchElementException;

import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.InitializingBean;

import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.vteba.utils.charstr.ByteUtils;
import com.vteba.utils.charstr.Char;
import com.vteba.utils.common.ClassUtils;

public class Kryoer implements InitializingBean {
	private static final Logger LOGGER = LogManager.getLogger(Kryoer.class);
	
	private List<Class<?>> classList;
	private ObjectPool<Kryo> objectPool;
	
	public Kryoer() {
		super();
	}

	public Kryoer(List<Class<?>> classList) {
		super();
		this.classList = classList;
		
	}
    
    /**
	 * 将对象序列化为byte[]，对象要注册
	 * @param object 要转换的对象
	 * @return 字节数组
	 * @author yinlei
	 * date 2013-4-6 下午2:51:28
	 */
	public byte[] toByte(Object object) {
	    if (object == null) {
            return null;
        }
	    Class<?> clazz = object.getClass();
	    Kryo kryo = borrowKryo();
        
		Output output = new Output(2048, 2 * 1024 * 1024);
		kryo.writeObject(output, object);
		
		String className = clazz.getName();
		byte[] nameBytes = className.getBytes(Char.UTF8);
		int length = nameBytes.length;
		
		byte[] objectByte = output.toBytes();
		int byteLength = objectByte.length;
		byte[] result = new byte[byteLength + 4 + length];
		
		ByteUtils.mergeBytes(length, result, 0, 3);
        System.arraycopy(nameBytes, 0, result, 4, length);
        System.arraycopy(objectByte, 0, result, length + 4, byteLength);
        
		returnKryo(kryo);
		
		return result;
	}

	/**
	 * 将字节数组反序列化为对象，对象要注册
	 * @param bytes 被转换的字节数组
	 * @return 要转换的对象实例
	 * @author yinlei
	 * date 2013-4-6 下午2:52:11
	 */
	public <T> T fromByte(byte[] bytes) {
	    if (bytes == null || bytes.length < 4) {
	        return null;
	    }
		
	    // className长度
		int length = getLength(bytes);
		
		String className = new String(bytes, 4, length, Char.UTF8);
		Class<T> clazz = ClassUtils.forName(className);
		
		Kryo kryo = borrowKryo();
		
		int targetLength = bytes.length - 4 -length;
		byte[] result = new byte[targetLength];
		System.arraycopy(bytes, 4 + length, result, 0, targetLength);
		
		Input input = new Input(result);
		T t = kryo.readObject(input, clazz);
		returnKryo(kryo);
		
		return t;
	}
	
	private int getLength(byte[] res) {
    	return (res[0] & 0xff) | ((res[1] << 8) & 0xff00) | ((res[2] << 24) >>> 8) | (res[3] << 24);
    }

	private void returnKryo(Kryo kryo) {
		try {
			objectPool.returnObject(kryo);
		} catch (Exception e) {
			LOGGER.error(e);
		}
	}

	private Kryo borrowKryo() {
		Kryo kryo = null;
		try {
			kryo = objectPool.borrowObject();
		} catch (NoSuchElementException e) {
			LOGGER.error("NoSuchElementException.", e);
		} catch (IllegalStateException e) {
			LOGGER.error("IllegalStateException.", e);
		} catch (Exception e) {
			LOGGER.error(e);
		}
		return kryo;
	}
	
	public List<Class<?>> getClassList() {
		return classList;
	}

	public void setClassList(List<Class<?>> classList) {
		this.classList = classList;
	}

	public ObjectPool<Kryo> getObjectPool() {
		return objectPool;
	}

	public void setObjectPool(ObjectPool<Kryo> objectPool) {
		this.objectPool = objectPool;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if (objectPool == null) {
			PooledObjectFactory<Kryo> pooledObjectFactory = new KryoPooledObjectFactory(classList);
			
			GenericObjectPoolConfig config = new GenericObjectPoolConfig();
			config.setMaxTotal(20);
			config.setTestOnBorrow(false);
	        config.setTestOnReturn(true);
	        config.setTestWhileIdle(true);
	        
	        objectPool = new GenericObjectPool<Kryo>(pooledObjectFactory, config);
		}
	}
}
